package com.sync.mutiParse;

import java.nio.ByteBuffer;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.middleware.race.sync.Server;
import com.constants.Constants;
import com.koloboke.collect.map.hash.HashIntIntMap;
import com.koloboke.collect.map.hash.HashIntIntMaps;

/**
 * 重放线程
 * 
 * @author LittleBearWinnie
 *
 */
public final class ReplayTask extends Thread {
	private static final Logger loggerServer = LoggerFactory
			.getLogger(Server.class);
	// ParkTask全部运行完，设置不生产
	public volatile static boolean producing = true;
	public int tag;// 线程tag
	public ByteBuffer middleBuffer;// 重放结果文件Buffer
	public HashIntIntMap indexMap = HashIntIntMaps.newMutableMap();
	public int indexPos = 4;// 防止 HashIntIntMap 的空位0

	public ReplayTask(int tag) {
		this.tag = tag;
	}

	@Override
	public void run() {
		loggerServer.info("重放线程开始运行");
		middleBuffer = ByteBuffer.allocate(Constants.reply_buff);// 可以改文件
		int readID = 0;
		Block block;
		int[] curPKs;// 1024-128
		int[] changePKs;
		int[] pos;
		byte[] types;
		byte[] buffer;
		int recordCount;
		try {
			while (true) {
				block = ReplayManager.replayMap.get(readID);
				if (block != null) {
					recordCount = block.recordCount;
					curPKs = block.curPKs;
					changePKs = block.changePKs;
					types = block.types;
					buffer = block.buffer;
					pos = block.pos;
					for (int i = 0; i < recordCount; ++i) {
						if (types[i] == Constants.UPDATE) {
							if (indexMap.containsKey(curPKs[i])) {
								int oldPos = indexMap.get(curPKs[i]);
								if (curPKs[i] != changePKs[i]) {
									indexMap.remove(curPKs[i]);
									indexMap.put(changePKs[i], oldPos);
								}
								GoodCodeUtils.handleUpdate(buffer, pos[i],
										middleBuffer, oldPos, curPKs[i]);
							}
						} else if (types[i] == Constants.INSERT) {
							if ((curPKs[i] & Constants.mod) == tag) {
								indexMap.put(curPKs[i], indexPos);
								GoodCodeUtils.handleInsert(buffer, pos[i],
										middleBuffer, indexPos, curPKs[i]);
								indexPos = middleBuffer.position();
							}
						} else {
							indexMap.remove(curPKs[i]);
						}
					}

					if (block.readTimes
							.incrementAndGet() >= Constants.ReplayThreadNum) {
						block.clear();
						 ReplayManager.replayMap.remove(readID);
					}
					readID++;
				} else if (producing) {// 优化
					TimeUnit.MICROSECONDS.sleep(1);
				} else {
					break;
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		// D|id:1:1|5817702|NULL|first_name:2:0|江|NULL|last_name:2:0|江|NULL|sex:2:0|男|NULL|score:1:0|55|NULL|
		loggerServer.info("中间存储大小: " + middleBuffer.position());
		loggerServer.info("重放线程结束,最后块数: " + readID);
	}
}
